package controller;

import operators.PV_Modify_01_AggregateFunction;
import operators.PV_Modify_01_KeyedProcessFunction;
import operators.PV_Modify_01_WindowedFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import pojo.PageViewCount;
import pojo.UserBehavior;

import java.net.URL;
import java.time.Duration;
import java.util.Random;

/*
* 2021-05-24 14:05:35
* 改进一： 解决以"pv"为keyBy导致的"热桶访问"问题，设置随机key，解决数据倾斜问题
* */
public class PageViewController_Modify_01 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(4);

        URL resource = PageViewController_Modify_01.class.getResource("/UserBehavior.csv");
        DataStream<String> fileDataStream = streamEnv.readTextFile(resource.getPath());

        DataStream<UserBehavior> watermarkDataStream = fileDataStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserBehavior>noWatermarks()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
        );

        KeyedStream<PageViewCount, Long> keyedDataStream = watermarkDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .map((MapFunction<UserBehavior, Tuple2<Integer, Integer>>) line ->
                        new Tuple2<>(new Random().nextInt(10), 1)) // 将元组Key值打散，keyBy分组后解决数据倾斜问题
                .returns(TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(Integer.class, Integer.class))
                .keyBy(value -> value.f0)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new PV_Modify_01_AggregateFunction(),
                        new PV_Modify_01_WindowedFunction())
                .keyBy(PageViewCount::getTimestamp);

        keyedDataStream.process(new PV_Modify_01_KeyedProcessFunction())
                .print();

        streamEnv.execute();
    }
}
